package com.spark.mooc.ch7_sparkstreaming.part05_outputOperation

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.spark.mooc.ch7_sparkstreaming.part02_basicInputSource.socket.StreamingExamples
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

/**
 * @description:
 * @time: 2020/11/30 21:23
 * @author: lhy
 */
object NetworkWordCountStatefulToMysql {
    def main(args: Array[String]): Unit = {
        // 定义状态更新函数
        def updateFunction = (values:Seq[Int], state:Option[Int]) => {
            val currentCount: Int = values.foldLeft(0)(_+_)
            val previousCount: Int = state.getOrElse(0)
            Some(currentCount + previousCount)
        }
        StreamingExamples.setStreamingLogLevels()       // 设置日志显示级别
        val conf: SparkConf = new SparkConf().setAppName("NetworkWordCountStateful").setMaster("local[2]")
        val ssc = new StreamingContext(conf,Seconds(5))
        ssc.checkpoint("output/streaming/checkpoint")
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.21.101",9999)
        val words: DStream[String] = lines.flatMap(_.split(" "))
        val wordDStream: DStream[(String, Int)] = words.map((_,1))
        val stateDStream: DStream[(String, Int)] = wordDStream.updateStateByKey[Int](updateFunction)
        stateDStream.print()
        /*
        把DStream保存到MySQL数据库中
         */
        stateDStream.foreachRDD(rdd => {
            // 内部函数
            def func(records: Iterator[(String,Int)]) {
                var conn :Connection = null
                var stmt: PreparedStatement = null
                try{
                    val url = "jdbc:mysql://192.168.21.104:3306/spark"
                    val user = "root"
                    val password = "bigdata"
                    conn = DriverManager.getConnection(url,user,password)
                    records.foreach(p => {
                        val sql = "insert into wordcount(word,count) values (?,?)"
                        stmt = conn.prepareStatement(sql)
                        stmt.setString(1,p._1.trim)
                        stmt.setInt(2,p._2.toInt)
                        stmt.executeUpdate()
                    })
                }catch {
                    case e: Exception => e.printStackTrace()
                }finally {
                    if (stmt != null)
                        stmt.close()
                    if (conn != null)
                        conn.close()
                }
            }
            val repartitionedRDD: RDD[(String, Int)] = rdd.repartition(3)
            repartitionedRDD.foreachPartition(func)
        })
        ssc.start()
        ssc.awaitTermination()
    }
}
